#if defined(OS_RTOS)
#include "aio-socket.h"
#include "list.h"
#include "sys/spinlock.h"
#include "sys/atomic.h"
#include <sys/pollfd.h>
#include <fcntl.h>
#include <errno.h>
#include <unistd.h>
#include <limits.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <assert.h>

#define POLL_MAX_FD 256

struct poll_model_list
{
	struct list_head link;
	int fd;
	int events;
	void *ctx;
};

struct model
{
	spinlock_t locker;
	struct list_head head;
};

static struct model s_poll;
static int s_threads = 0;

typedef union poll_data
{
  void *ptr;
  int fd;
  uint32_t u32;
  uint64_t u64;
} poll_data_t;

struct poll_event {
	uint32_t events;
	poll_data_t data;
};

struct poll_context_accept
{
	aio_onaccept proc;
	void *param;
};

struct poll_context_connect
{
	aio_onconnect proc;
	void *param;
	struct sockaddr_storage addr;  // for send to
	socklen_t addrlen;
};

struct poll_context_recv
{
	aio_onrecv proc;
	void *param;
	void *buffer;
	size_t bytes;
};

struct poll_context_send
{
	aio_onsend proc;
	void *param;
	const void *buffer;
	size_t bytes;
	struct sockaddr_storage peer;  // for send to
	socklen_t peerlen;
	struct sockaddr_storage local;  // for sendmsg
	socklen_t locallen;
};

struct poll_context_recv_v
{
	aio_onrecv proc;
	void *param;
	socket_bufvec_t *vec;
	int n;
};

struct poll_context_send_v
{
	aio_onsend proc;
	void *param;
	socket_bufvec_t *vec;
	int n;
	struct sockaddr_storage peer;  // for send to
	socklen_t peerlen;
	struct sockaddr_storage local;  // for sendmsg
	socklen_t locallen;
};

struct poll_context_recvfrom
{
	aio_onrecvfrom proc;
	void *param;
	void *buffer;
	size_t bytes;
};

struct poll_context_recvfrom_v
{
	aio_onrecvfrom proc;
	aio_onrecvmsg proc2;
	void *param;
	socket_bufvec_t *vec;
	int n;
};

struct poll_context
{
	spinlock_t locker; // memory alignment, see more about Apple Developer spinlock
	struct poll_event ev[2];
	socket_t socket[2];
	volatile int32_t ref;
	int own;
	struct list_head list;
	bool destroy;

	aio_ondestroy ondestroy;
	void* param;

	int (*read)(struct poll_context *ctx, int flags, int code);
	int (*write)(struct poll_context *ctx, int flags, int code);

	union
	{
		struct poll_context_accept accept;
		struct poll_context_recv recv;
		struct poll_context_recv_v recv_v;
		struct poll_context_recvfrom recvfrom;
		struct poll_context_recvfrom_v recvfrom_v;
	} in;
	socket_bufvec_t vec[2][1]; // for recvmsg/sendmsg

	union
	{
		struct poll_context_connect connect;
		struct poll_context_send send;
		struct poll_context_send_v send_v;
	} out;
};

#define PollCtrl(ctx, flag, idx) do {                            \
			struct poll_model_list* model = poll_model_create(); \
			if (NULL == model) return -1;                        \
			atomic_increment32(&ctx->ref);                       \
			spinlock_lock(&ctx->locker);                         \
			ctx->ev[idx].events |= flag;                         \
			model->ctx    = ctx;                                 \
			model->events = ctx->ev[idx].events;                 \
			model->fd     = ctx->socket[idx];                    \
			spinlock_unlock(&ctx->locker);                       \
			spinlock_lock(&s_poll.locker);                       \
			list_insert_after(&model->link, s_poll.head.prev);   \
			spinlock_unlock(&s_poll.locker);                     \
			return 0;                                            \
		} while (0)

#define PollIn(ctx, callback)	ctx->read = callback; PollCtrl(ctx, POLLIN, 0)
#define PollOut(ctx, callback)	ctx->write = callback; PollCtrl(ctx, POLLOUT, 1)

struct poll_model_list* poll_model_create(void)
{
	struct poll_model_list* model;

	model = (struct poll_model_list*)calloc(1, sizeof(*model));
	if (model)
	{
		LIST_INIT_HEAD(&model->link);
	}
	return model;
}

int poll_model_release(struct poll_model_list* model)
{
	free(model);
	return 0;
}

static int aio_socket_release(struct poll_context* ctx)
{
	if (0 == atomic_decrement32(&ctx->ref))
	{
		if(ctx->own)
			close(ctx->socket[0]);

		spinlock_destroy(&ctx->locker);

		if (ctx->ondestroy)
			ctx->ondestroy(ctx->param);

#if defined(DEBUG) || defined(_DEBUG)
		memset(ctx, 0xCC, sizeof(*ctx));
#endif
		free(ctx);
	}
	return 0;
}

int aio_socket_init(int threads)
{
	s_threads = threads;
	LIST_INIT_HEAD(&s_poll.head);
	spinlock_create(&s_poll.locker);
	return 0;
}

int aio_socket_clean(void)
{
	spinlock_destroy(&s_poll.locker);
	return 0;
}

int aio_socket_process(int timeout)
{
	int i, r = 0, count = 0;
	uint32_t userevent;
	struct poll_context *ctx;
	struct list_head *pos, *next;
	struct poll_model_list *model;
	struct pollfd *fds = (struct pollfd *)calloc(POLL_MAX_FD, sizeof(*fds));
	if (NULL == fds) return -1;

	spinlock_lock(&s_poll.locker);
	list_for_each_safe(pos, next, &s_poll.head) {
		model = list_entry(pos, struct poll_model_list, link);
		fds[count].fd = model->fd;
		fds[count].events = model->events | POLLERR | POLLNVAL;
		fds[count].revents = 0;
		count++;
		if (count >= POLL_MAX_FD) break;
	}
	spinlock_unlock(&s_poll.locker);
	if (count == 0)
		goto exit;

	r = poll(fds, count, timeout);
	if (0 == r || r < 0)
		goto exit;

	for(i = 0; i < count; i++)
	{
		if (0 == fds[i].revents) continue;

		// POLLERR: some poll error occurred
		// POLLNVAL: the specified file descriptor is invalid
		int flags = POLLERR | POLLNVAL;
		userevent = 0;
		ctx = NULL;

		spinlock_lock(&s_poll.locker);
		list_for_each_safe(pos, next, &s_poll.head) {
			model = list_entry(pos, struct poll_model_list, link);
			if (fds[i].fd == model->fd && ((((struct poll_context*)model->ctx)->ev[0].events | fds[i].revents) || 
				(((struct poll_context*)model->ctx)->ev[1].events | fds[i].revents))) {
				ctx = (struct poll_context*)model->ctx;
				list_remove(&model->link);
				poll_model_release(model);
				model = NULL;
				break;
			}
		}
		spinlock_unlock(&s_poll.locker);
		if (NULL == ctx) continue;
		assert(NULL != ctx);
		assert(ctx->ref > 0);

		if(fds[i].revents & flags)
		{
			// save event
			spinlock_lock(&ctx->locker);
			if (ctx->ev[0].events & POLLIN)
			{
				userevent = ctx->ev[0].events;
				ctx->ev[0].events &= ~(POLLIN | POLLOUT);
			}
			else if (ctx->ev[1].events & POLLOUT)
			{
				userevent = ctx->ev[1].events;
				ctx->ev[1].events &= ~(POLLIN | POLLOUT);
			}
			spinlock_unlock(&ctx->locker);

			// error
			if(POLLIN & userevent)
			{
				assert(ctx->read);
				ctx->read(ctx, 1, EPIPE);
				aio_socket_release(ctx);
			}

			if(POLLOUT & userevent)
			{
				assert(ctx->write);
				ctx->write(ctx, 1, EPIPE);
				aio_socket_release(ctx);
			}
		}
		else
		{
			spinlock_lock(&ctx->locker);
			if ((fds[i].revents & POLLIN) & ctx->ev[0].events)
			{
				userevent = POLLIN;
				ctx->ev[0].events &= ~POLLIN;
			}
			else if ((fds[i].revents & POLLOUT) & ctx->ev[1].events)
			{
				userevent = POLLOUT;
				ctx->ev[1].events &= ~POLLOUT;
			}
			spinlock_unlock(&ctx->locker);

			if(POLLIN & userevent)
			{
				assert(ctx->read);
				ctx->read(ctx, 1, 0);
				aio_socket_release(ctx);
			}
			else if(POLLOUT & userevent)
			{
				assert(ctx->write);
				ctx->write(ctx, 1, 0);
				aio_socket_release(ctx);
			}
			else if(fds[i].revents & (POLLIN|POLLOUT))
			{
				// nothing to do
				aio_socket_release(ctx);
			}
		}
		r = 1;
		break;
	}

exit:
	if (fds) {
		free(fds);
		fds = NULL;
	}
	return r;
}

aio_socket_t aio_socket_create(socket_t socket, int own)
{
//	int flags;
	struct poll_context* ctx;
	ctx = (struct poll_context*)calloc(1, sizeof(struct poll_context));
	if(!ctx)
		return NULL;

	spinlock_create(&ctx->locker);
	ctx->own = own;
	ctx->ref = 1; // 1-for aio_socket_destroy, not support POLLHUP
	ctx->socket[0] = socket;
	ctx->ev[0].data.ptr = ctx;

	ctx->socket[1] = socket;
	ctx->ev[1].data.ptr = ctx;

	// set non-blocking socket, for Edge Triggered
	//flags = fcntl(socket, F_GETFL, 0);
	//fcntl(socket, F_SETFL, flags | O_NONBLOCK);

	return ctx;
}

int aio_socket_destroy(aio_socket_t socket, aio_ondestroy ondestroy, void* param)
{
	bool find;
	uint32_t userevent = 0;
	struct list_head *pos, *next;
	struct poll_model_list *model;
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(ctx->ev[0].data.ptr == ctx);
	ctx->ondestroy = ondestroy;
	ctx->param = param;

	shutdown(ctx->socket[0], SHUT_RDWR);
//	close(sock[0]); // can't close socket now, avoid socket reuse

retry:
	find = false;
	spinlock_lock(&s_poll.locker);
	list_for_each_safe(pos, next, &s_poll.head) {
		model = list_entry(pos, struct poll_model_list, link);
		if(model->ctx == ctx) {
			list_remove(&model->link);
			poll_model_release(model);
			find = true;
		}
	}
	spinlock_unlock(&s_poll.locker);

	if (find) {
		// save event
		spinlock_lock(&ctx->locker);
		if (ctx->ev[0].events & POLLIN)
		{
			userevent = ctx->ev[0].events;
			ctx->ev[0].events &= ~(POLLIN | POLLOUT);
		}
		else if (ctx->ev[1].events & POLLOUT)
		{
			userevent = ctx->ev[1].events;
			ctx->ev[1].events &= ~(POLLIN | POLLOUT);
		}
		spinlock_unlock(&ctx->locker);

		// error
		if(POLLIN & userevent)
		{
			assert(ctx->read);
			ctx->read(ctx, 1, EPIPE);
			aio_socket_release(ctx);
		}

		if(POLLOUT & userevent)
		{
			assert(ctx->write);
			ctx->write(ctx, 1, EPIPE);
			aio_socket_release(ctx);
		}
		goto retry;
	}

	aio_socket_release(ctx);
	return 0;
}

static int poll_accept(struct poll_context* ctx, int flags, int error)
{
	socket_t client;
	struct sockaddr_storage addr;
	socklen_t addrlen = sizeof(addr);

	if(0 != error)
	{
		assert(1 == flags); // only in poll_wait thread
		ctx->in.accept.proc(ctx->in.accept.param, error, 0, NULL, 0);
		return error;
	}

	client = accept(ctx->socket[0], (struct sockaddr*)&addr, &addrlen);
	if(client > 0)
	{
		ctx->in.accept.proc(ctx->in.accept.param, 0, client, (struct sockaddr*)&addr, addrlen);
		return 0;
	}
	else
	{
		assert(-1 == client);
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->in.accept.proc(ctx->in.accept.param, errno, 0, NULL, 0);
		return 0;
	}
}

int aio_socket_accept(aio_socket_t socket, aio_onaccept proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[0].events & POLLIN));
	if(ctx->ev[0].events & POLLIN)
		return EBUSY;

	ctx->in.accept.proc = proc;
	ctx->in.accept.param = param;

	// man 2 accept to see more
	PollIn(ctx, poll_accept);
	return errno; // poll_ctl return -1
}

static int poll_connect(struct poll_context* ctx, int flags, int error)
{
	socklen_t len;

    // call in poll_wait thread
    assert(1 == flags);

    if(0 != error)
	{
		ctx->out.connect.proc(ctx->out.connect.param, error);
		return error;
	}
    else
    {
        // man connect to see more (EINPROGRESS)
		len = sizeof(error);
        getsockopt(ctx->socket[1], SOL_SOCKET, SO_ERROR, (void*)&error, &len);
        ctx->out.connect.proc(ctx->out.connect.param, error);
        return error;
    }
}

int aio_socket_connect(aio_socket_t socket, const struct sockaddr *addr, socklen_t addrlen, aio_onconnect proc, void* param)
{
	int r;
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[1].events & POLLOUT));
	if(ctx->ev[1].events & POLLOUT)
		return EBUSY;

	ctx->out.connect.addrlen = addrlen > sizeof(ctx->out.connect.addr) ? sizeof(ctx->out.connect.addr) : addrlen;
	memcpy(&ctx->out.connect.addr, addr, ctx->out.connect.addrlen);
	ctx->out.connect.proc = proc;
	ctx->out.connect.param = param;

    r = connect(ctx->socket[1], (const struct sockaddr*)&ctx->out.connect.addr, ctx->out.connect.addrlen);
    if(0 == r || EINPROGRESS == errno)
    {
        PollOut(ctx, poll_connect);
    }
	return errno; // poll_ctl return -1
}

static int poll_recv(struct poll_context* ctx, int flags, int error)
{
	ssize_t r;

	r = recv(ctx->socket[0], ctx->in.recv.buffer, ctx->in.recv.bytes, 0);
	if(r >= 0)
	{
		ctx->in.recv.proc(ctx->in.recv.param, 0, (size_t)r);
		return 0;
	}
	else
	{
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->in.recv.proc(ctx->in.recv.param, errno, 0);
		return 0;
	}
}

int aio_socket_recv(aio_socket_t socket, void* buffer, size_t bytes, aio_onrecv proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[0].events & POLLIN));
	if(ctx->ev[0].events & POLLIN)
		return EBUSY;

	ctx->in.recv.proc = proc;
	ctx->in.recv.param = param;
	ctx->in.recv.buffer = buffer;
	ctx->in.recv.bytes = bytes;

	PollIn(ctx, poll_recv);
	return errno; // poll_ctl return -1
}

static int poll_send(struct poll_context* ctx, int flags, int error)
{
	ssize_t r;
	if(0 != error)
	{
		assert(1 == flags); // only in poll_wait thread
		ctx->out.send.proc(ctx->out.send.param, error, 0);
		return error;
	}

	r = send(ctx->socket[1], ctx->out.send.buffer, ctx->out.send.bytes, 0);
	if(r >= 0)
	{
		ctx->out.send.proc(ctx->out.send.param, 0, (size_t)r);
		return 0;
	}
	else
	{
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->out.send.proc(ctx->out.send.param, errno, 0);
		return 0;
	}
}

int aio_socket_send(aio_socket_t socket, const void* buffer, size_t bytes, aio_onsend proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[1].events & POLLOUT));
	if(ctx->ev[1].events & POLLOUT)
		return EBUSY;

	ctx->out.send.proc = proc;
	ctx->out.send.param = param;
	ctx->out.send.buffer = buffer;
	ctx->out.send.bytes = bytes;

	PollOut(ctx, poll_send);
	return errno; // poll_ctl return -1
}

static int poll_recv_v(struct poll_context* ctx, int flags, int error)
{
	ssize_t r;
	struct msghdr msg;

	memset(&msg, 0, sizeof(msg));
	msg.msg_iov = (struct iovec*)ctx->in.recv_v.vec;
	msg.msg_iovlen = ctx->in.recv_v.n;

	r = recvmsg(ctx->socket[0], &msg, 0);
	if(r >= 0)
	{
		ctx->in.recv_v.proc(ctx->in.recv_v.param, 0, (size_t)r);
		return 0;
	}
	else
	{
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->in.recv_v.proc(ctx->in.recv_v.param, errno, 0);
		return 0;
	}
}

int aio_socket_recv_v(aio_socket_t socket, socket_bufvec_t* vec, int n, aio_onrecv proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[0].events & POLLIN));
	if(ctx->ev[0].events & POLLIN)
		return EBUSY;

	ctx->in.recv_v.proc = proc;
	ctx->in.recv_v.param = param;
	ctx->in.recv_v.vec = vec;
	ctx->in.recv_v.n = n;

	PollIn(ctx, poll_recv_v);
	return errno; // poll_ctl return -1
}

static int poll_send_v(struct poll_context* ctx, int flags, int error)
{
	ssize_t r;
	struct msghdr msg;

	if(0 != error)
	{
		assert(1 == flags); // only in poll_wait thread
		ctx->out.send_v.proc(ctx->out.send_v.param, error, 0);
		return error;
	}

	memset(&msg, 0, sizeof(msg));
	msg.msg_iov = (struct iovec*)ctx->out.send_v.vec;
	msg.msg_iovlen = ctx->out.send_v.n;

	r = sendmsg(ctx->socket[1], &msg, 0);
	if(r >= 0)
	{
		ctx->out.send_v.proc(ctx->out.send_v.param, 0, (size_t)r);
        return 0;
	}
	else
	{
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->out.send_v.proc(ctx->out.send_v.param, errno, 0);
		return 0;
	}
}

int aio_socket_send_v(aio_socket_t socket, socket_bufvec_t* vec, int n, aio_onsend proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[1].events & POLLOUT));
	if(ctx->ev[1].events & POLLOUT)
		return EBUSY;

	ctx->out.send_v.proc = proc;
	ctx->out.send_v.param = param;
	ctx->out.send_v.vec = vec;
	ctx->out.send_v.n = n;

	PollOut(ctx, poll_send_v);
	return errno; // poll_ctl return -1
}

static int poll_recvfrom(struct poll_context* ctx, int flags, int error)
{
	ssize_t r;
	struct sockaddr_storage addr;
	socklen_t addrlen = sizeof(addr);

	if(0 != error)
	{
		assert(1 == flags); // only in poll_wait thread
		ctx->in.recvfrom.proc(ctx->in.recvfrom.param, error, 0, NULL, 0);
		return error;
	}

	r = recvfrom(ctx->socket[0], ctx->in.recvfrom.buffer, ctx->in.recvfrom.bytes, 0, (struct sockaddr*)&addr, &addrlen);
	if(r >= 0)
	{
		ctx->in.recvfrom.proc(ctx->in.recvfrom.param, 0, (size_t)r, (struct sockaddr*)&addr, addrlen);
		return 0;
	}
	else
	{
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->in.recvfrom.proc(ctx->in.recvfrom.param, errno, 0, NULL, 0);
		return 0;
	}
}

int aio_socket_recvfrom(aio_socket_t socket, void* buffer, size_t bytes, aio_onrecvfrom proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[0].events & POLLIN));
	if(ctx->ev[0].events & POLLIN)
		return EBUSY;

	ctx->in.recvfrom.proc = proc;
	ctx->in.recvfrom.param = param;
	ctx->in.recvfrom.buffer = buffer;
	ctx->in.recvfrom.bytes = bytes;

	PollIn(ctx, poll_recvfrom);
	return errno; // poll_ctl return -1
}

static int poll_sendto(struct poll_context* ctx, int flags, int error)
{
	ssize_t r;
	if(0 != error)
	{
		assert(1 == flags); // only in poll_wait thread
		ctx->out.send.proc(ctx->out.send.param, error, 0);
		return error;
	}

	r = sendto(ctx->socket[1], ctx->out.send.buffer, ctx->out.send.bytes, 0, (struct sockaddr*)&ctx->out.send.peer, ctx->out.send.peerlen);
	if(r >= 0)
	{
		ctx->out.send.proc(ctx->out.send.param, 0, (size_t)r);
		return 0;
	}
	else
	{
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->out.send.proc(ctx->out.send.param, errno, 0);
		return 0;
	}
}

int aio_socket_sendto(aio_socket_t socket, const struct sockaddr *addr, socklen_t addrlen, const void* buffer, size_t bytes, aio_onsend proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[1].events & POLLOUT));
	if(ctx->ev[1].events & POLLOUT)
		return EBUSY;

	ctx->out.send.peerlen = addrlen > sizeof(ctx->out.send.peer) ? sizeof(ctx->out.send.peer) : addrlen;
	memcpy(&ctx->out.send.peer, addr, ctx->out.send.peerlen);
	ctx->out.send.proc = proc;
	ctx->out.send.param = param;
	ctx->out.send.buffer = buffer;
	ctx->out.send.bytes = bytes;

	PollOut(ctx, poll_sendto);
	return errno; // poll_ctl return -1
}

static int poll_recvfrom_v(struct poll_context* ctx, int flags, int error)
{
	ssize_t r;
	char control[64];
	struct msghdr hdr;
	struct cmsghdr *cmsg;
	socklen_t locallen;
	struct sockaddr_storage peer, local;

	if(0 != error)
	{
		assert(1 == flags); // only in poll_wait thread
		ctx->in.recvfrom_v.proc ? ctx->in.recvfrom_v.proc(ctx->in.recvfrom_v.param, error, 0, NULL, 0) :
									ctx->in.recvfrom_v.proc2(ctx->in.recvfrom_v.param, error, 0, NULL, 0, NULL, 0);
		return error;
	}

	memset(&peer, 0, sizeof(peer));
	memset(&hdr, 0, sizeof(hdr));
	memset(control, 0, sizeof(control));
	hdr.msg_name = &peer;
	hdr.msg_namelen = sizeof(peer);
	hdr.msg_iov = (struct iovec*)ctx->in.recvfrom_v.vec;
	hdr.msg_iovlen = ctx->in.recvfrom_v.n;
	hdr.msg_control = control;
	hdr.msg_controllen = sizeof(control);
	hdr.msg_flags = 0;

	r = recvmsg(ctx->socket[0], &hdr, 0);
	if(r >= 0)
	{
		locallen = 0;
		memset(&local, 0, sizeof(local));
		for (cmsg = CMSG_FIRSTHDR(&hdr); !!cmsg && ctx->in.recvfrom_v.proc2; cmsg = CMSG_NXTHDR(&hdr, cmsg))
		{
			if (cmsg->cmsg_level == IPPROTO_IP && cmsg->cmsg_type == IP_PKTINFO)
			{
				struct in_pktinfo* pktinfo;
				pktinfo = (struct in_pktinfo*)CMSG_DATA(cmsg);
				locallen = sizeof(struct sockaddr_in);
				((struct sockaddr_in*)&local)->sin_family = AF_INET;
				memcpy(&((struct sockaddr_in*)&local)->sin_addr, &pktinfo->ipi_addr, sizeof(pktinfo->ipi_addr));
				break;
			}
#if !defined(OS_RTOS) && defined(_GNU_SOURCE)
			else if (cmsg->cmsg_level == IPPROTO_IPV6 && cmsg->cmsg_type == IPV6_PKTINFO)
			{
				struct in6_pktinfo* pktinfo6;
				pktinfo6 = (struct in6_pktinfo*)CMSG_DATA(cmsg);
				locallen = sizeof(struct sockaddr_in6);
				((struct sockaddr_in6*)&local)->sin6_family = AF_INET6;
				memcpy(&((struct sockaddr_in6*)&local)->sin6_addr, &pktinfo6->ipi6_addr, sizeof(pktinfo6->ipi6_addr));
				break;
			}
#endif
		}

		ctx->in.recvfrom_v.proc ? ctx->in.recvfrom_v.proc(ctx->in.recvfrom_v.param, 0, (size_t)r, (struct sockaddr*)&peer, hdr.msg_namelen) :
									ctx->in.recvfrom_v.proc2(ctx->in.recvfrom_v.param, 0, (size_t)r, (struct sockaddr*)&peer, hdr.msg_namelen, (struct sockaddr*)&local, locallen);
		return 0;
	}
	else
	{
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->in.recvfrom_v.proc ? ctx->in.recvfrom_v.proc(ctx->in.recvfrom_v.param, errno, 0, NULL, 0) :
									ctx->in.recvfrom_v.proc2(ctx->in.recvfrom_v.param, errno, 0, NULL, 0, NULL, 0);
		return 0;
	}
}

int aio_socket_recvfrom_v(aio_socket_t socket, socket_bufvec_t* vec, int n, aio_onrecvfrom proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[0].events & POLLIN));
	if(ctx->ev[0].events & POLLIN)
		return EBUSY;

	ctx->in.recvfrom_v.proc = proc;
	ctx->in.recvfrom_v.proc2 = NULL;
	ctx->in.recvfrom_v.param = param;
	ctx->in.recvfrom_v.vec = vec;
	ctx->in.recvfrom_v.n = n;

	PollIn(ctx, poll_recvfrom_v);
	return errno; // poll_ctl return -1
}

static int poll_sendto_v(struct poll_context* ctx, int flags, int error)
{
	ssize_t r;
	char control[64];
	struct msghdr hdr;
	struct cmsghdr* cmsg;
	struct sockaddr_storage* local;

	if(0 != error)
	{
		assert(1 == flags); // only in poll_wait thread
		ctx->out.send_v.proc(ctx->out.send_v.param, error, 0);
		return error;
	}

	memset(&hdr, 0, sizeof(hdr));
	memset(control, 0, sizeof(control));
	hdr.msg_name = (struct sockaddr*)&ctx->out.send_v.peer;
	hdr.msg_namelen = ctx->out.send_v.peerlen;
	hdr.msg_iov = (struct iovec*)ctx->out.send_v.vec;
	hdr.msg_iovlen = ctx->out.send_v.n;
	hdr.msg_control = control;
	hdr.msg_controllen = sizeof(control);
	hdr.msg_flags = 0;

	cmsg = CMSG_FIRSTHDR(&hdr);
	local = &ctx->out.send_v.local;
	if (AF_INET == local->ss_family && ctx->out.send_v.locallen >= sizeof(struct sockaddr_in))
	{
		struct in_pktinfo* pktinfo;
		cmsg->cmsg_level = IPPROTO_IP; // SOL_IP
		cmsg->cmsg_type = IP_PKTINFO;
		cmsg->cmsg_len = CMSG_LEN(sizeof(struct in_pktinfo));
		pktinfo = (struct in_pktinfo*)CMSG_DATA(cmsg);
		memset(pktinfo, 0, sizeof(struct in_pktinfo));
#if defined(OS_RTOS)
		memcpy(&pktinfo->ipi_addr, &((struct sockaddr_in*)local)->sin_addr, sizeof(pktinfo->ipi_addr));
#else
		memcpy(&pktinfo->ipi_spec_dst, &((struct sockaddr_in*)local)->sin_addr, sizeof(pktinfo->ipi_spec_dst));
#endif
		hdr.msg_controllen = CMSG_SPACE(sizeof(struct in_pktinfo));
	}
#if !defined(OS_RTOS) && defined(_GNU_SOURCE)
	else if (AF_INET6 == local->ss_family && ctx->out.send_v.locallen >= sizeof(struct sockaddr_in6))
	{
		struct in6_pktinfo* pktinfo6; // struct ipv6_mreq
		cmsg->cmsg_level = IPPROTO_IPV6;
		cmsg->cmsg_type = IPV6_PKTINFO;
		cmsg->cmsg_len = CMSG_LEN(sizeof(struct in6_pktinfo));
		pktinfo6 = (struct in6_pktinfo*)CMSG_DATA(cmsg);
		memset(pktinfo6, 0, sizeof(struct in6_pktinfo));
		memcpy(&pktinfo6->ipi6_addr, &((struct sockaddr_in6*)local)->sin6_addr, sizeof(pktinfo6->ipi6_addr));
		hdr.msg_controllen = CMSG_SPACE(sizeof(struct in6_pktinfo));
	}
#endif
	else
	{
		hdr.msg_controllen = 0;
	}

	r = sendmsg(ctx->socket[1], &hdr, 0);
	if(r >= 0)
	{
		ctx->out.send_v.proc(ctx->out.send_v.param, 0, (size_t)r);
		return 0;
	}
	else
	{
		if(0 == flags)
			return errno;

		// call in poll_wait thread
		ctx->out.send_v.proc(ctx->out.send_v.param, errno, 0);
		return 0;
	}
}

int aio_socket_sendto_v(aio_socket_t socket, const struct sockaddr *addr, socklen_t addrlen, socket_bufvec_t* vec, int n, aio_onsend proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[1].events & POLLOUT));
	if(ctx->ev[1].events & POLLOUT)
		return EBUSY;

	ctx->out.send_v.locallen = 0;
	ctx->out.send_v.peerlen = addrlen > sizeof(ctx->out.send_v.peer) ? sizeof(ctx->out.send_v.peer) : addrlen;
	memcpy(&ctx->out.send_v.peer, addr, ctx->out.send_v.peerlen);
	ctx->out.send_v.proc = proc;
	ctx->out.send_v.param = param;
	ctx->out.send_v.vec = vec;
	ctx->out.send_v.n = n;

	PollOut(ctx, poll_sendto_v);
	return errno; // poll_ctl return -1
}

int aio_socket_recvmsg(aio_socket_t socket, void* buffer, size_t bytes, aio_onrecvmsg proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	ctx->vec[0][0].iov_base = buffer;
	ctx->vec[0][0].iov_len = bytes;
	return aio_socket_recvmsg_v(socket, ctx->vec[0], 1, proc, param);
}

int aio_socket_sendmsg(aio_socket_t socket, const struct sockaddr* peer, socklen_t peerlen, const struct sockaddr* local, socklen_t locallen, const void* buffer, size_t bytes, aio_onsend proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	ctx->vec[1][0].iov_base = (void*)buffer;
	ctx->vec[1][0].iov_len = bytes;
	return aio_socket_sendmsg_v(socket, peer, peerlen, local, locallen, ctx->vec[1], 1, proc, param);
}

int aio_socket_recvmsg_v(aio_socket_t socket, socket_bufvec_t* vec, int n, aio_onrecvmsg proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[0].events & POLLIN));
	if (ctx->ev[0].events & POLLIN)
		return EBUSY;

	ctx->in.recvfrom_v.proc = NULL;
	ctx->in.recvfrom_v.proc2 = proc;
	ctx->in.recvfrom_v.param = param;
	ctx->in.recvfrom_v.vec = vec;
	ctx->in.recvfrom_v.n = n;

	PollIn(ctx, poll_recvfrom_v);
	return errno; // poll_ctl return -1
}

int aio_socket_sendmsg_v(aio_socket_t socket, const struct sockaddr* peer, socklen_t peerlen, const struct sockaddr* local, socklen_t locallen, socket_bufvec_t* vec, int n, aio_onsend proc, void* param)
{
	struct poll_context* ctx = (struct poll_context*)socket;
	assert(0 == (ctx->ev[1].events & POLLOUT));
	if (ctx->ev[1].events & POLLOUT)
		return EBUSY;

	ctx->out.send_v.peerlen = peerlen > sizeof(ctx->out.send_v.peer) ? sizeof(ctx->out.send_v.peer) : peerlen;
	memcpy(&ctx->out.send_v.peer, peer, ctx->out.send_v.peerlen);
	ctx->out.send_v.locallen = locallen > sizeof(ctx->out.send_v.local) ? sizeof(ctx->out.send_v.local) : locallen;
	memcpy(&ctx->out.send_v.local, local, ctx->out.send_v.locallen);
	ctx->out.send_v.proc = proc;
	ctx->out.send_v.param = param;
	ctx->out.send_v.vec = vec;
	ctx->out.send_v.n = n;

	PollOut(ctx, poll_sendto_v);
	return errno; // poll_ctl return -1
}

#endif